package com.jnpc.spark

import com.jnpc.spark.project.utils.{CRProcess, FamenUtil, NLPUtil}
import org.apache.spark.sql.SparkSession


/**
  * Created by yangqiang on 2018/3/28 0028.
  */
object BatchProcessDataOnYARN2 {
  def main(args: Array[String]): Unit = {

    if (args.length != 1) {
      println("Usage: BatchProcessDataOnYARN2 <folderpath>  ")
      System.exit(1)
    }
    val Array(folderpath) = args
    val spark = SparkSession.builder()
      .appName("BatchProcessDataOnYARNApp2")
      .master("local[*]")
      .getOrCreate()
    val crlistcnnpRDD = spark.sparkContext.textFile(folderpath + "crreport")
    val calistcnnpRDD = spark.sparkContext.textFile(folderpath + "calistcnnp")
    val iercapreportRDD = spark.sparkContext.textFile(folderpath + "iercapreport")
    val loerreportRDD = spark.sparkContext.textFile(folderpath + "loerreport")
    /*  val crlistcnnpRDD = spark.sparkContext.textFile("D:\\DEV\\NLP\\LTP\\dealdata\\crreport")
       val calistcnnpRDD = spark.sparkContext.textFile("D:\\DEV\\NLP\\LTP\\dealdata\\calistcnnp")
       val iercapreportRDD = spark.sparkContext.textFile("D:\\DEV\\NLP\\LTP\\dealdata\\iercapreport")
       val loerreportRDD = spark.sparkContext.textFile("D:\\DEV\\NLP\\LTP\\dealdata\\loerreport")*/
    crlistcnnpRDD.map(_.split("@"))
      .map(line =>
        CRReport(line(0), line(1), line(2), line(3), line(4), line(5),
          line(6), line(7), line(8))
      ).foreach(cr=>{
      new CRProcess().CreateCRreport(cr)
    })

     iercapreportRDD.map(_.split("@"))
      .map(line =>
        IER_CAP_CNNP(line(0), line(1), line(2), line(3), line(4))
      ).foreach(ier=>{
       new CRProcess().CreateIERCAPCNNP(ier)
     })

     loerreportRDD.map(_.split("@"))
      .map(line =>
        LOERREPORT(line(0), line(1), line(2), line(3), line(4), line(5), line(6), line(7), line(8), line(9), line(10))
      ).foreach(loer=>{
       new CRProcess().CreateLOERREPORT(loer)
     })

    calistcnnpRDD.map(_.split("@"))
      .map(line =>
        CA_LIST_CNNP(line(0), line(1), line(2), line(3), line(4), line(5))
      ).foreach(ca=>{
      new CRProcess().CreateCAList(ca)
    })
    spark.stop
  }


  case class CRReport(id_key_cr: String, cr_no: String, cr_equipment: String,
                      cr_unit: String, cr_date: String, cr_unit_stat: String,
                      cr_subject: String, cr_desc: String, cr_cause: String)

  case class LOERREPORT(id_key_cr: String, loer_no: String, haf_no: String,
                        unit_code: String, uniform_unit: String, report_year: String,
                        report_serial: String, t_subject: String, t_start_event: String,
                        event_begin: String, event_end: String)

  case class IER_CAP_CNNP(id_key_cr: String, er_no: String, cnnp_ioer_no: String,
                          t_er_name: String, t_er_desc: String)

  case class CA_LIST_CNNP(id_key_cr: String, id_key_ca: String, report_no: String,
                          ca_no: String, trace_no: String, t_ca_desc: String)

}
